package org.ykhl.mq.commons.receiver;

import com.google.common.base.Strings;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import org.ykhl.mq.commons.BaseMessage;
import org.ykhl.mq.commons.ConnectionException;
import org.ykhl.mq.commons.Constants;
import org.ykhl.mq.commons.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ykhl.test.XnLog;

import java.util.concurrent.BlockingQueue;

/**
 * 消息接收类的抽象父类，用于维护一些基本的生命周期管理逻辑和通用的成员变量等。
 * 
 * @author Zhang Kefei
 *
 */
public abstract class MessageReceiver {
	private static final Logger LOG = LoggerFactory.getLogger(MessageReceiver.class);

	private static class CacheHolder {
		private static final Cache<String, BlockingQueue<BaseMessage>> RESPONSE_QUEUE_CACHE =
				CacheBuilder.newBuilder().recordStats().weakValues().build();
	}
	private static final Cache<String, BlockingQueue<BaseMessage>> getResponseQueueCache() {
		return CacheHolder.RESPONSE_QUEUE_CACHE;
	}
	public static final void putQueueCache(String reqId, BlockingQueue<BaseMessage> queue) {
		getResponseQueueCache().put(reqId, queue);
	}
	// used as the secret key for app server
	protected String authKey;
	protected String topic;
	protected String consumerGroupName;
	protected Constants.Role role;
	private volatile MessageListener listener;

	// the server address of MQ, usually name server address
	protected String serverAddress;
	protected Boolean useSsl;
	protected String securityFilePath;

	private boolean running = false;

	protected MessageReceiver(String serverAddress, String authKey, String securityFilePath,
							  Constants.Role role, MessageListener listener) {

		init(serverAddress, authKey, securityFilePath, role,
				Constants.MQ_TO_APPSERVER_TOPIC_PREFIX,
				Constants.MQ_TO_APPSERVER_CONSUMER_PREFIX, listener);
	}

	protected MessageReceiver(String serverAddress, String authKey, String securityFilePath,
							  Constants.Role role, String topicPrefix,
							  String consumerPrefix, MessageListener listener){
		init(serverAddress, authKey, securityFilePath, role, topicPrefix, consumerPrefix, listener);
	}

	private void init(String serverAddress, String authKey, String securityFilePath,
					  Constants.Role role, String topicPrefix,
					  String consumerPrefix, MessageListener listener){

		if (Strings.isNullOrEmpty(authKey)) {
			if (Strings.isNullOrEmpty(securityFilePath)) {
				throw new NullPointerException(
						"either authKey or securityFilePath should be non-null");
			}

			authKey = Utils.getAuthIdFromKeyStoreUnchecked(securityFilePath);
		}

		if(Strings.isNullOrEmpty(serverAddress) || Strings.isNullOrEmpty(authKey))
			throw new NullPointerException("serverAddress or authKey is empty");

		this.serverAddress = serverAddress;
		this.authKey = authKey;
		this.securityFilePath = Strings.emptyToNull(securityFilePath);
		this.listener = listener;
		this.role = role;

		topic = topicPrefix + authKey;
		consumerGroupName = consumerPrefix + authKey;
	}

	/**
	 * 设置推送消息的消息监听器.
	 *
	 * @param listener
     */
	public void setListener(MessageListener listener) {
		this.listener = listener;
	}

	/**
	 * 设置MessageReceiver连接的服务器地址。
	 *
	 * @param serverAddress
     */
	public void setServerAddress(String serverAddress){
		if(serverAddress == null)
			throw new NullPointerException();

		this.serverAddress = serverAddress;
	}

	/**
	 * 设置是否使用加密及认证模式进行传输。
	 *
	 * @param useSsl
     */
	public void setUseSsl(Boolean useSsl) {
		this.useSsl = useSsl;
	}

	public boolean isRunning(){
		return running;
	}

	/**
	 * 用于启动消息接收器的方法，调用前请确保listener已经设置。
	 *
	 * @throws ConnectionException
	 */
	public void start() throws ConnectionException {
		if(running)		// already running
			return;
		if(listener == null)	// listener not set, can't consume messages for now
			throw new NullPointerException("listener not set.");

		try{
			doStart();

			running = true;
		} catch(Exception e){
			throw new ConnectionException(e);
		}
	}

	/**
	 * 用于停止消息接收器的方法。
	 *
	 * @throws ConnectionException
	 */
	public void stop() throws ConnectionException {
		if(!running)
			return;

		try{
			doStop();

			running = false;
		} catch(Exception e){
			throw new ConnectionException(e);
		}
	}

	protected MessageConsumeStatus consume(BaseMessage baseMessage){
		long start = System.currentTimeMillis();
//		XnLog.toLogInfo("Cacheinfo hitCount: " + CacheHolder.RESPONSE_QUEUE_CACHE.stats().hitCount()
//				+ "--- evictionCount: " + CacheHolder.RESPONSE_QUEUE_CACHE.stats().evictionCount()
//				+ "--- hitRate: " + CacheHolder.RESPONSE_QUEUE_CACHE.stats().hitRate()
//				+ "--- averageLoadPenalty: " + CacheHolder.RESPONSE_QUEUE_CACHE.stats().averageLoadPenalty()
//				+ "--- missCount: " + CacheHolder.RESPONSE_QUEUE_CACHE.stats().missCount()
//				+ "--- loadCount: " + CacheHolder.RESPONSE_QUEUE_CACHE.stats().loadCount()
//				+ "--- loadExceptionCount: " + CacheHolder.RESPONSE_QUEUE_CACHE.stats().loadExceptionCount()
//				+ "--- loadExceptionRate: " + CacheHolder.RESPONSE_QUEUE_CACHE.stats().loadExceptionRate()
//				+ "--- loadSuccessCount: " + CacheHolder.RESPONSE_QUEUE_CACHE.stats().loadSuccessCount()
//				+ "--- missRate: " + CacheHolder.RESPONSE_QUEUE_CACHE.stats().missRate()
//				+ "--- requestCount: " + CacheHolder.RESPONSE_QUEUE_CACHE.stats().requestCount()
//				+ "--- totalLoadTime: " + CacheHolder.RESPONSE_QUEUE_CACHE.stats().totalLoadTime()
//		);

		String requestId = baseMessage.getRequestId();
		BlockingQueue<BaseMessage> queue;
		if (requestId != null &&
				(queue = getResponseQueueCache().getIfPresent(requestId)) != null){
			queue.add(baseMessage);
			XnLog.toLog("consume queue add 消耗时间："+(System.currentTimeMillis() -start));
		} else{
			XnLog.toLog("consume listener consume requestId："+ requestId+"-- queue："+getResponseQueueCache().getIfPresent(requestId));
			XnLog.toLog("consume listener consume 消耗时间："+(System.currentTimeMillis() -start));
			MessageConsumeStatus status = listener.consume(baseMessage);
			return status;
		}
		return MessageConsumeStatus.CONSUME_SUCCESS;
	}
	
	protected abstract void doStart() throws Exception;
	protected abstract void doStop() throws Exception;
}
